package com.ysy.mq.db;


import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.google.common.util.concurrent.RateLimiter;
import com.ysy.constant.RocketMqConstant;
import com.ysy.entity.dto.NoteCollectDBDTO;
import com.ysy.entity.dto.NoteLikeStatusCountDTO;
import com.ysy.mapper.NoteCountMapper;
import com.ysy.mapper.UserCountMapper;
import com.ysy.persimmons.common.util.StringUtils;
import jakarta.annotation.Resource;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.CollectionUtils;

import java.util.List;


@Component
@RocketMQMessageListener(consumerGroup = "persimmons_count_group_"+RocketMqConstant.NOTE_COLLECT_COUNT_DB_TOPIC,
        topic = RocketMqConstant.NOTE_COLLECT_COUNT_DB_TOPIC)
public class NoteCollectCountDBConsumer implements RocketMQListener<String> {

    private static Logger logger = LoggerFactory.getLogger(NoteCollectCountDBConsumer.class);

    @Resource
    private NoteCountMapper noteCountmapper;
    @Resource
    private UserCountMapper userCountMapper;
    @Resource
    private TransactionTemplate transactionTemplate;

    private RateLimiter rateLimiter = RateLimiter.create(5000);

    @Override
    public void onMessage(String message) {

        if (StringUtils.isBlank(message)) {
            return;
        }

        // acquire() 会阻塞线程，直到获取到 permit。
        // 如果不想阻塞，可以用 tryAcquire()（非阻塞，立即返回是否成功）
        rateLimiter.acquire();

        List<NoteCollectDBDTO> countDTOList = JSON.parseObject(message, new TypeReference<List<NoteCollectDBDTO>>() {});
        if (CollectionUtils.isEmpty(countDTOList)) {
            return;
        }
        for (NoteCollectDBDTO countDTO : countDTOList) {
            Integer count = countDTO.getCount();
            transactionTemplate.execute(status->{
                try {
                    noteCountmapper.updateOrInsertCollectCount(countDTO.getNoteId(),count);
                    userCountMapper.updateOrInsertCollectCount(countDTO.getNoteCreatorId(),count);
                    return true;
                } catch (Exception ex) {
                    status.setRollbackOnly();
                    logger.error("点赞统计存储DB失败：{}，具体异常信息：",countDTO ,ex);
                }
                return false;
            });
        }

    }
}
